package com.ruoyi.quartz.task;

import cn.hutool.json.JSONUtil;
import com.ruoyi.common.utils.exception.AjaxException;
import com.ruoyi.system.domain.SysOrder;
import com.ruoyi.system.domain.SysUser;
import com.ruoyi.system.domain.SysUserAccount;
import com.ruoyi.system.domain.SysUserBillcash;
import com.ruoyi.system.mapper.SysUserAccountMapper;
import com.ruoyi.system.mapper.SysUserBillcashMapper;
import com.ruoyi.system.mapper.SysUserMapper;
import com.ruoyi.system.service.other.MessageEntity;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import com.ruoyi.common.utils.StringUtils;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Date;
import java.util.List;
import java.util.Map;


/**
 * 定时任务调度测试
 * 
 * @author ruoyi
 */
@Component("ryTask")
public class RyTask
{
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private SysUserAccountMapper sysUserAccountMapper;

    @Autowired
    private SysUserBillcashMapper sysUserBillcashMapper;

    @Autowired
    private SysUserMapper sysUserMapper;


    public void ryMultipleParams(String s, Boolean b, Long l, Double d, Integer i)
    {
        System.out.println(StringUtils.format("执行多参方法： 字符串类型{}，布尔类型{}，长整型{}，浮点型{}，整形{}", s, b, l, d, i));
    }

    public void ryParams(String params)
    {
        System.out.println("执行有参方法：" + params);
    }

    @Transactional
    public void ryNoParams()
    {
        StreamOperations<String, String, String> streamOperations = this.stringRedisTemplate.opsForStream();

/*        // 从消费者的pending队列中读取消息
        List<MapRecord<String, String, String>>  retVal = streamOperations.read(Consumer.from("group-1", "consumer_1"), StreamOffset.create("mystream", ReadOffset.from("0")));

        // 遍历消息
        for (MapRecord<String, String, String> record : retVal ) {
            // 消费消息
            LOGGER.info("消息value="+record.getValue());
            // 手动ack消息
            streamOperations.acknowledge("my_group", record);
        }*/
        // 获取my_group中的pending消息信息，本质上就是执行XPENDING指令
        PendingMessagesSummary pendingMessagesSummary = streamOperations.pending("mystream", "group-1");
        // 所有pending消息的数量
        long totalPendingMessages = pendingMessagesSummary.getTotalPendingMessages();

        // 消费组名称
        String groupName= pendingMessagesSummary.getGroupName();


        // pending队列中的最小ID
        String minMessageId;
        // pending队列中的最大ID
        String maxMessageId;
        try {
            minMessageId = pendingMessagesSummary.minMessageId();
            maxMessageId = pendingMessagesSummary.maxMessageId();
        }catch (Exception e){
            System.out.println("pending列表无数据");
            return;
        }

        // pending队列中的最大ID

        // 每个消费者的pending消息数量
        Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer();

        pendingMessagesPerConsumer.entrySet().forEach(entry -> {

            // 消费者
            String consumer = entry.getKey();
            // 消费者的pending消息数量
            long consumerTotalPendingMessages = entry.getValue();


            if (consumerTotalPendingMessages > 0 && consumer.equals("consumer-1")) {
                // 读取消费者pending队列的前10条记录，从ID=0的记录开始，一直到ID最大值
                PendingMessages pendingMessages = streamOperations.pending("mystream", Consumer.from("group-1", consumer), Range.closed("0", "+"), 100);

                // 遍历所有Opending消息的详情
                pendingMessages.forEach(message -> {
                    // 消息的ID
                    RecordId recordId =  message.getId();
                    // 消息从消费组中获取，到此刻的时间
                    Duration elapsedTimeSinceLastDelivery = message.getElapsedTimeSinceLastDelivery();
                    // 消息被获取的次数
                    long deliveryCount = message.getTotalDeliveryCount();


                    /**
                     * 演示手动消费的这个判断非常的针对，目的就是要读取消费者“my_consumer1”pending消息中，ID=1605524665215-0的这条消息
                     */
                    // 通过streamOperations，直接读取这条pending消息，
                    List<MapRecord<String, String, String>> result = streamOperations.range("mystream", Range.rightOpen(recordId.toString(), recordId.toString()));

                    // 开始和结束都是同一个ID，所以结果只有一条
                    MapRecord<String, String, String> record = result.get(0);

                    // 这里执行日志输出，模拟的就是消费逻辑
                    MessageEntity messageEntity = JSONUtil.toBean(record.getValue().get("name"),MessageEntity.class);
                    if (messageEntity.getKey().equals("pay")){
                        //支付回调
                        SysOrder sysOrder = JSONUtil.toBean(messageEntity.getContent(),SysOrder.class);
                        SysUserAccount sysUserAccount = sysUserAccountMapper.selectSysUserAccountById(sysOrder.getUserId());
                        if (sysUserAccount==null){
                            throw new AjaxException("account invalid");
                        }
                        BigDecimal oldBalance = sysUserAccount.getBalance();
                        BigDecimal newBalance = sysUserAccount.getBalance().add(sysOrder.getOrderLastAmount());
                        sysUserAccount.setBalance(newBalance);

                        SysUserBillcash sysUserBillcash = new SysUserBillcash();
                        sysUserBillcash.setRidCode(sysOrder.getOrderCode());
                        sysUserBillcash.setChannelId(sysOrder.getChannelId());
                        sysUserBillcash.setChannelName(sysOrder.getChannelName());
                        sysUserBillcash.setAmount(oldBalance);
                        sysUserBillcash.setChangeAmount(sysOrder.getOrderLastAmount());
                        sysUserBillcash.setLastAmount(newBalance);
                        SysUser sysUser = sysUserMapper.selectUserById(sysOrder.getUserId());
                        if (sysUser==null){
                            throw new AjaxException("user error");
                        }
                        sysUserBillcash.setUserId(sysUser.getUserId());
                        sysUserBillcash.setUserCode(sysUser.getUserCode());
                        sysUserBillcash.setLoginName(sysUser.getLoginName());
                        sysUserBillcash.setBillcashType(0);
                        sysUserBillcash.setBillcashRemark("支付完成");
                        sysUserBillcash.setCreateTime(new Date());
                        sysUserBillcash.setUpdateTime(new Date());
                        try {
                            sysUserBillcashMapper.insertSysUserBillcash(sysUserBillcash);
                        }catch (Exception e){
                            throw new AjaxException("system error:mysql");
                        }

                        //更新账户失败 放回队列
                        try {
                            sysUserAccountMapper.updateSysUserAccount(sysUserAccount);
                        }catch (Exception e){
                            throw new AjaxException("system error:account");
                        }
                    }else {
                        //代付回调
                    }

                    // 如果手动消费成功后，往消费组提交消息的ACK
                    Long retVal = streamOperations.acknowledge("group-1", record);
                });
            }
        });
//        System.out.println("执行无参方法");
    }
}
